package nacos_grpc

import (
	"context"
	"errors"
	"fmt"
	"gitee.com/xlizy/common-go/base/common_config"
	"gitee.com/xlizy/common-go/base/common_goroutine"
	"gitee.com/xlizy/common-go/base/enums/common_error"
	"gitee.com/xlizy/common-go/components/nacos/v2"
	"gitee.com/xlizy/common-go/components/orm/tx"
	"gitee.com/xlizy/common-go/utils/common_utils"
	"gitee.com/xlizy/common-go/utils/json"
	"gitee.com/xlizy/common-go/utils/threadlocal"
	"gitee.com/xlizy/common-go/utils/zlog"
	"github.com/google/uuid"
	"github.com/nacos-group/nacos-sdk-go/v2/model"
	"github.com/nacos-group/nacos-sdk-go/v2/vo"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/connectivity"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/grpc/metadata"
	"google.golang.org/grpc/status"
	"net"
	"strings"
	"sync"
	"time"
)

var (
	CtxOptionsTimeOut = "timeout"
	CtxOptionsShowLog = "showLog"
	traceIdKey        = "trace_id"
)

var consumerConnsOnce sync.Once
var consumerConns = make(map[string]map[string]*grpc.ClientConn)
var consumerConnsLock = sync.RWMutex{}

var availabilityCluster = []string{"DEFAULT"}

type Service struct {
	Name      string
	Weight    float64
	RegHandle func(s grpc.ServiceRegistrar)
}

type Services struct {
	Consumer []Service
	Provider []Service
}

type localService struct {
	Name   string
	Port   uint64
	Server *grpc.Server
}

var localServices = make([]localService, 0)

var providerInterceptor = func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
	showLog := true
	if ctx.Value(CtxOptionsShowLog) != nil {
		showLog = ctx.Value(CtxOptionsShowLog).(bool)
	}
	if showLog {

	}
	defer func() {
		if _err := recover(); _err != nil {
			if tx.IsInTX() {
				tx.NeedRollback()
			}
			errMsg := fmt.Sprintf("%s", _err)
			zlog.Error("Grpc接口实现方法发生异常:{},{}", info.FullMethod, errMsg)
			err = status.Error(codes.Code(common_error.RpcExecuteError.Code()), common_error.RpcExecuteError.Des())
		}
		if tx.IsInTX() {
			tx.TryCommit()
		}
	}()

	md, ok := metadata.FromIncomingContext(ctx)
	if ok {
		var traceId string
		value, ok := md[traceIdKey]
		if ok {
			traceId = value[0]
		} else {
			traceId = uuid.New().String()
		}
		threadlocal.SetTraceId(traceId)
	}

	if showLog {
		zlog.Info("grpc请求体:{}", json.ToJsonStr(req))
	}
	resp, err = handler(ctx, req)
	if err != nil {
		zlog.Error("grpc调用方法发生异常:{}", err.Error())
	}
	if showLog {
		zlog.Info("grpc结果:{}", json.ToJsonStr(resp))
	}
	return resp, err
}

var consumerInterceptor = func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
	showLog := true
	timeOut := 30 * time.Second
	if ctx.Value(CtxOptionsShowLog) != nil {
		showLog = ctx.Value(CtxOptionsShowLog).(bool)
	}
	if ctx.Value(CtxOptionsTimeOut) != nil {
		timeOut = ctx.Value(CtxOptionsTimeOut).(time.Duration)
	}
	if showLog {
		zlog.Info("调用grpc接口method:{},conn:{}", method, cc.Target())
	}

	timeOutCtx, cancel := context.WithTimeout(ctx, timeOut)
	defer cancel()
	timeOutCtx = metadata.NewOutgoingContext(timeOutCtx, metadata.Pairs(traceIdKey, threadlocal.GetTraceId()))
	if showLog {
		zlog.Info("grpc请求体:{}", json.ToJsonStr(req))
	}
	err := invoker(timeOutCtx, method, req, reply, cc, opts...)
	if showLog {
		zlog.Info("grpc结果:{}", json.ToJsonStr(reply))
	}
	if tx.IsInTX() {
		if err != nil {
			tx.NeedRollback()
		}
	}
	return err
}

func InitNacosGrpc(services Services) {
	cfg := common_config.GetNacosCfg()
	availabilityCluster = strings.Split(cfg.AvailabilityCluster, ",")
	namingClient := nacos.GetNamingClient()
	if namingClient == nil {
		zlog.Error("连接Nacos异常:namingClient is nil")
		panic("连接Nacos异常")
	} else {
		for _, service := range services.Provider {
			goroutine_service := service
			port, _ := common_utils.GetFreePort()
			listen, err1 := net.Listen("tcp", fmt.Sprintf(":%d", port))
			if err1 != nil {
				zlog.Error("监听TCP端口异常", err1.Error())
				panic(err1)
			} else {
				server := grpc.NewServer(grpc.UnaryInterceptor(providerInterceptor), grpc.MaxRecvMsgSize(1024*1024*50))
				_ = common_goroutine.GetNonBlockingAntsPool().Submit(func() {
					goroutine_service.RegHandle(server)
					if err := server.Serve(listen); err != nil {
						zlog.Error("启动grpc服务异常:{}", err.Error())
						panic("启动grpc服务异常")
					}
				})
				localServices = append(localServices, localService{
					Name:   service.Name,
					Port:   uint64(port),
					Server: server,
				})
				regOk, err2 := namingClient.RegisterInstance(vo.RegisterInstanceParam{
					Ip:          common_utils.GetLocalPriorityIp(common_config.PriorityNetwork.Networks),
					Port:        uint64(port),
					ServiceName: "grpc:" + service.Name,
					Weight:      service.Weight,
					Enable:      true,
					Healthy:     true,
					Ephemeral:   true,
					GroupName:   "DEFAULT_GROUP",
				})
				if err2 != nil {
					zlog.Error("注册grpc服务异常:{}", err2.Error())
					panic("注册grpc服务异常")
				} else {
					if !regOk {
						zlog.Error("注册grpc服务异常失败")
					}
				}
			}
		}

		var updateConsumerConns = func(serviceName string, instances []model.Instance) {
			//zlog.Info("updateConsumerConns...{},{}", serviceName, json.ToJsonStr(instances))
			consumerConnsLock.Lock()
			defer consumerConnsLock.Unlock()

			if len(instances) == 0 {
				s, e1 := namingClient.GetService(vo.GetServiceParam{
					ServiceName: "grpc:" + serviceName,
					GroupName:   "DEFAULT_GROUP",
				})
				if e1 == nil {
					instances = s.Hosts
				}
			}

			//zlog.Info("instances:{}", json.ToJsonStr(instances))

			delDelConsumerConnsKeys := make(map[string]string)
			for connKey, _ := range consumerConns[serviceName] {
				delDelConsumerConnsKeys[connKey] = connKey
			}

			for _, instance := range instances {
				connKey := instance.Ip + ":" + fmt.Sprintf("%d", instance.Port)
				delete(delDelConsumerConnsKeys, connKey)

				if instance.Healthy && instance.Enable {
					if conn, connExist := consumerConns[serviceName][connKey]; connExist {
						//如果连接存在，并且状态不可用；关连接、删缓存
						if conn != nil && (conn.GetState() == connectivity.TransientFailure || conn.GetState() == connectivity.Shutdown) {
							_ = conn.Close()
							delete(consumerConns[serviceName], connKey)

							zlog.Info("将新增Grpc连接serviceName:{},connKey:{}", serviceName, connKey)
							newConn, newConnErr := grpc.NewClient(fmt.Sprintf("%s:%d", instance.Ip, instance.Port), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithUnaryInterceptor(consumerInterceptor))
							if newConnErr != nil {
								zlog.Error("连接grpc服务异常:{}", newConnErr.Error())
							} else {
								consumerConns[serviceName][connKey] = newConn
							}
						}
					} else {
						zlog.Info("将新增Grpc连接serviceName:{},connKey:{}", serviceName, connKey)
						newConn, newConnErr := grpc.NewClient(fmt.Sprintf("%s:%d", instance.Ip, instance.Port), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithUnaryInterceptor(consumerInterceptor))
						if newConnErr != nil {
							zlog.Error("连接grpc服务异常:{}", newConnErr.Error())
						} else {
							consumerConns[serviceName][connKey] = newConn
						}
					}
				} else {
					if conn, connExist := consumerConns[serviceName][connKey]; connExist {
						//如果连接存在，并且状态不可用；关连接、删缓存
						if conn != nil && (conn.GetState() == connectivity.TransientFailure || conn.GetState() == connectivity.Shutdown) {
							_ = conn.Close()
						}
						delete(consumerConns[serviceName], connKey)
					}
				}
			}

			//删除、释放无用连接
			for key, _ := range delDelConsumerConnsKeys {
				if conn, connExist := consumerConns[serviceName][key]; connExist {
					if conn != nil {
						_ = conn.Close()
						delete(consumerConns[serviceName], key)
					}
				}
			}

			//zlog.Info("最新Grpc连接信息,serviceName:{},consumerConns:{}", serviceName, json.ToJsonStr(consumerConns))
		}

		for _, service := range services.Consumer {
			goroutineService := service
			_, ok := consumerConns[goroutineService.Name]
			if !ok {
				consumerConns[goroutineService.Name] = make(map[string]*grpc.ClientConn)
			}
			instances, err1 := namingClient.SelectAllInstances(vo.SelectAllInstancesParam{
				ServiceName: "grpc:" + goroutineService.Name,
				GroupName:   "DEFAULT_GROUP",
			})
			if err1 != nil {
				zlog.Error("获取grpc可用实例列表异常:{}", err1.Error())
			} else {
				updateConsumerConns(goroutineService.Name, instances)
			}
			err0 := nacos.GetNamingClient().Subscribe(&vo.SubscribeParam{
				ServiceName: "grpc:" + goroutineService.Name,
				GroupName:   "DEFAULT_GROUP",
				SubscribeCallback: func(services []model.Instance, err error) {
					threadlocal.SetTraceId(uuid.New().String())
					zlog.Info("收到Nacos服务订阅消息service:{},Instances:{}", goroutineService.Name, json.ToJsonStr(services))
					updateConsumerConns(goroutineService.Name, services)
				},
			})
			if err0 != nil {
				zlog.Error("订阅grpc服务异常:{}", err0.Error())
			}

			_, _ = nacos.GetCronManage().AddFunc("0/30 * * * * *", func() {
				i, e2 := namingClient.SelectAllInstances(vo.SelectAllInstancesParam{
					ServiceName: "grpc:" + goroutineService.Name,
					GroupName:   "DEFAULT_GROUP",
				})
				if e2 == nil {
					updateConsumerConns(goroutineService.Name, i)
				} else {
					zlog.Info("SelectAllInstances异常:{}", e2.Error())
				}
			})

		}
	}
}

func GetGrpcConn(serviceName string) *grpc.ClientConn {
	var conn *grpc.ClientConn
	connKeyExist := false
	connKey, err := nacos.GetAppIns("grpc:" + serviceName)
	zlog.Info("GetAppIns:{}", connKey)
	if err == nil {
		consumerConnsLock.RLock()
		conn, connKeyExist = consumerConns[serviceName][connKey]
		consumerConnsLock.RUnlock()
		if connKeyExist {
			return conn
		} else {
			for k, v := range consumerConns {
				zlog.Info("缓存连接,k:{},v:{}", k, v)
			}
			zlog.Error("获取grpc服务实例异常-未找到可用实例:{}", serviceName)
			panic(errors.New("无可用Grpc服务:" + serviceName))
		}
	} else {
		zlog.Error("无可用Grpc服务", err)
		panic(errors.New("无可用Grpc服务:" + serviceName))
	}
	return conn
}

// GracefulShutdown 优雅关闭
func GracefulShutdown() {
	client := nacos.GetNamingClient()
	if client != nil {
		for _, service := range localServices {
			_, _ = client.DeregisterInstance(vo.DeregisterInstanceParam{
				Ip:          common_utils.GetLocalPriorityIp(common_config.PriorityNetwork.Networks),
				Port:        service.Port,
				ServiceName: "grpc:" + service.Name,
				Ephemeral:   true,
				GroupName:   "DEFAULT_GROUP",
			})
			if service.Server != nil {
				service.Server.GracefulStop()
			}
		}
	}
}
